Media Artifacts Interface of verl-omni
Pinned verl-omni version 45fd746. This is the specification for the media output of diffusion models.
Table of Contents
1. Motivation
Before PR#557, verl-omni has to infer what a tensor is from its shape and return type. For example, take ndim==5 tensors as videos; check if a tensor is SD3’s latent by checking if one dimension is \(16\); distinguish videos and audios generated by LTX2 from tuple using convention.
The core idea of PR#557 is to create a record data structure and ask adapter explicitly returns what each output is, so that downstreams can extract what they need from named fields from this record data structure, rather than guessing.
2. Solution
2.1. Contract Layer: Data Abstraction
2.1.1. Declaration MediaSpec
@dataclass(frozen=True)
class MediaSpec:
modality: Literal["image", "video", "audio"]
representation: Literal["decoded", "latent"]
# For example "CHW" / "TCHW" / "CT" / "LC" / "CTHW"
layout: str
sample_rate: int | None = None
fps: float | None = None
MediaSpec models the metadata of a single tensor, along with its semantic meanings, including shape interpretation, modality, decoded or latent, etc. Based on MediaSpec, we can model all outputs of a certain model, using dictionary that maps a string to a MediaSpec.
@dataclass(frozen=True)
class DiffusionIOSpec:
"""Available named artifacts, with canonical decoded and native latent axes.
Each request selects its primary and optional preview explicitly. Runtime
sample rate/FPS live on returned artifacts; declarations constrain only the
adapter-owned artifact names, representations, and layouts.
"""
artifacts: Mapping[str, MediaSpec]
def __post_init__(self) -> None:
if not isinstance(self.artifacts, Mapping) or not self.artifacts:
raise ValueError("DiffusionIOSpec requires named artifacts")
if any(not isinstance(name, str) or not isinstance(spec, MediaSpec) for name, spec in self.artifacts.items()):
raise TypeError("DiffusionIOSpec requires artifact-name to MediaSpec declarations")
The DiffusionIOSpec maps a name to the corresponding semantic tensor, through artifacts field.
2.1.2. Runtime Object MediaArtifact
MediaArtifact(spec, data, context) packs the metadata, the corresponding tensor, and context, which records pipeline and request_id for downstream debugging.
@dataclass(frozen=True)
class MediaArtifact:
"""A named stream's declaration and one sample's tensor (no batch axis)."""
spec: MediaSpec
data: torch.Tensor
context: str = ""
def validate(self, *, context: str, name: str) -> None: pass
def normalized(self, *, context: str, name: str) -> MediaArtifact: pass
It has 2 methods. The validate() method is to check the validity of the metadata and tensor; the normalized() method is to modify the underlying tensor to the required shape.